package com.cxl.mapreduce._15topN;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * 讲输入文档的总流量按照从高到低取前10 总流量为最后一个字段
 * 输入内容：
 * 13470253144	180	180	360
 * 13509468723	7335	110349	117684
 * 13560439638	918	4938	5856
 * 13568436656	3597	25635	29232
 * 13590439668	1116	954	2070
 * 13630577991	6960	690	7650
 * 13682846555	1938	2910	4848
 * 13729199489	240	0	240
 * 13736230513	2481	24681	27162
 * 13768778790	120	120	240
 * 13846544121	264	0	264
 * 13956435636	132	1512	1644
 * 13966251146	240	0	240
 * 13975057813	11058	48243	59301
 * 13992314666	3008	3720	6728
 * 15043685818	3659	3538	7197
 * 15910133277	3156	2936	6092
 * 15959002129	1938	180	2118
 * 18271575951	1527	2106	3633
 * 18390173782	9531	2412	11943
 * 84188413	4116	1432	5548
 * 输出内容：
 * 13509468723	7335	110349	117684
 * 13975057813	11058	48243	59301
 * 13568436656	3597	25635	29232
 * 13736230513	2481	24681	27162
 * 18390173782	9531	2412	11943
 * 13630577991	6960	690	7650
 * 15043685818	3659	3538	7197
 * 13992314666	3008	3720	6728
 * 15910133277	3156	2936	6092
 * 13560439638	918	4938	5856
 */
public class TopNDriver {
	public static void main(String[] args) throws IllegalArgumentException, IOException, ClassNotFoundException, InterruptedException {

		// 输入输出路径需要根据自己电脑上实际的输入输出路径设置
		args = new String[] { "/Users/a123/devWorkspace/bigData/hdfs/src/main/resources/input/topNInput.txt",
				"/Users/a123/devWorkspace/bigData/hdfs/src/main/resources/output/topN" };

		// 1 获取配置信息，或者job对象实例
		Configuration configuration = new Configuration();
		Job job = Job.getInstance(configuration);

		// 6 指定本程序的jar包所在的本地路径
		job.setJarByClass(TopNDriver.class);

		// 2 指定本业务job要使用的mapper/Reducer业务类
		job.setMapperClass(TopNMapper.class);
		job.setReducerClass(TopNReducer.class);

		// 3 指定mapper输出数据的kv类型
		job.setMapOutputKeyClass(FlowBean.class);
		job.setMapOutputValueClass(Text.class);

		// 4 指定最终输出的数据的kv类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(FlowBean.class);

		// 5 指定job的输入原始文件所在目录
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		// 7 将job中配置的相关参数，以及job所用的java类所在的jar包， 提交给yarn去运行
		boolean result = job.waitForCompletion(true);
		System.exit(result ? 0 : 1);

	}
}
